package com.jaymin.flink.datastream.source;

import com.jaymin.flink.datastream.utils.MysqlUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @author jaymin
 * @since 2024/4/5 19:56
 */
public class MysqlSource extends RichSourceFunction<Student> {
    Connection connection;
    PreparedStatement preparedStatement;

    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        ResultSet resultSet = preparedStatement.executeQuery();
        while (resultSet.next()) {
            int id = resultSet.getInt("id");
            String name = resultSet.getString("name");
            int age = resultSet.getInt("age");
            Student student = new Student(id, name, age);
            ctx.collect(student);
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = MysqlUtils.getConnection();
        preparedStatement = connection.prepareStatement("select * from student");
    }

    @Override
    public void cancel() {
        MysqlUtils.close(connection, preparedStatement);
    }
}
